package com.flute.haflute.jobbox.communication;

import java.net.MalformedURLException;
import java.rmi.Naming;
import java.rmi.Remote;
import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.Hashtable;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.flute.haflute.jobbox.common.CloudConstants;
import com.flute.haflute.jobbox.common.CloudContext;
import com.flute.haflute.jobbox.master.JobClient;

/**
 * 主节点通信平台接口实现
 * 
 */
public class CommunicationServiceFactoryImpl extends UnicastRemoteObject
		implements CommunicationServiceFactory,
		SpecificCommunicationInterface4Master {

	private static Logger logger = LoggerFactory
			.getLogger(CommunicationServiceFactoryImpl.class);

	private static final long serialVersionUID = 1L;

	private static volatile CommunicationServiceFactoryImpl instance = null;

	/** RMI监听端口 */
	private int port = 1099;

	/** 主节点的地址 */
	private String host = "localhost";

	// 所有消息管理器列表
	private Map<String, Remote> serviceFactory;

	// 所有同步消息管理器列表
	private Notifier notifier = new Notifier();

	private JobClient jobClient = new JobClient();

	public static CommunicationServiceFactoryImpl getInstance()
			throws RemoteException {
		if (instance == null) {
			synchronized (CommunicationServiceFactoryImpl.class) {
				if (instance == null)
					instance = new CommunicationServiceFactoryImpl();
			}
		}
		return instance;
	}

	private CommunicationServiceFactoryImpl() throws RemoteException {
		serviceFactory = new Hashtable<String, Remote>();
		// host = CloudContext.getLocalIP();
		port = CloudContext.getRMIPort();
		rebind();
	}

	/**
	 * 向RMI服务器注册
	 * 
	 * @throws RemoteException
	 */
	private void rebind() throws RemoteException {

		try {
			// 绑定 COMMUNICATION SERVER
			String ServerMoudle = "rmi://" + host + ":" + port + "/"
					+ CloudConstants.COMMUNICATION_SERV_CENTER;

			logger.info("starting RMI service: {}", ServerMoudle);

			Naming.rebind(ServerMoudle, this);
			logger.info(" {} started", ServerMoudle);

			// 绑定JOB SUBMITTOR, FOR CLIENT TO USE
			ServerMoudle = "rmi://" + host + ":" + port + "/"
					+ CloudConstants.JOB_SUBMITTER;

			logger.info("starting RMI service: {}", ServerMoudle);

			Naming.rebind(ServerMoudle, jobClient);
			logger.info(" {} started", ServerMoudle);
		} catch (Exception e) {
			logger.error("start RMI service failed ", e);
			System.err.println("bind rmi service on port=" + port + " failed");
			throw new RemoteException(e.getMessage());
		}
	}

	@Override
	public Remote getMessager(String messagerName) throws RemoteException {
		Remote result = serviceFactory.get(messagerName);
		if (result == null)
			throw new RemoteException("service not found :" + messagerName);
		return result;
	}

	@Override
	public void putMessager(String messagerName, Remote messager)
			throws RemoteException {
		serviceFactory.put(messagerName, messager);
	}

	@Override
	public synchronized Remote removeMessager(String messagerName) {
		return serviceFactory.remove(messagerName);
	}

	@Override
	public NotifyService getNotifier() throws RemoteException {
		return notifier;
	}

	@Override
	public int getDaemonNo(String ip) throws RemoteException {
		return CloudContext.getChildDaemonNo(ip);
	}

	@Override
	public void bindService(String serviceName, Remote service)
			throws RemoteException, MalformedURLException {

		String ServerMoudle = "rmi://" + host + ":" + port + "/" + serviceName;
		logger.info("starting RMI service: {}", ServerMoudle);

		Naming.rebind(ServerMoudle, service);
		logger.info(" {} started", ServerMoudle);

	}

	@Override
	public void setConfiguredNumDaemons(String nodeIp, int numDaemons)
			throws RemoteException {
		CloudContext.setNumDaemons4Node(nodeIp, numDaemons);
	}

}
